package com.shujia;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

// 加载电信数据到HBase中，使用HBase提供的多版本特性
// 根据用户mdn 获取用户最新10个位置记录
public class Demo3DianXin {
    Connection conn = null;
    Admin admin = null;
    Table dianxin = null;
    TableName name = TableName.valueOf("dianxin");

    @Before
    public void init() {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");
        try {
            conn = ConnectionFactory.createConnection(conf);
            admin = conn.getAdmin();
            dianxin = conn.getTable(name);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Test
    public void createTable() throws IOException {
        if (!admin.tableExists(name)) {
            admin.createTable(new HTableDescriptor(name).addFamily(new HColumnDescriptor("cf1").setMaxVersions(10)));
        } else {
            System.out.println("表已存在");
        }
    }

    @Test
    public void loadData() throws IOException {
        BufferedReader br = new BufferedReader(new FileReader("data/DIANXIN.csv"));
        String line = null;
        while ((line = br.readLine()) != null) {
            String[] splits = line.split(",");
            String mdn = splits[0];
            long start_time = Long.parseLong(splits[1]);
            String longitude = splits[4];
            String latitude = splits[5];
            Put put = new Put(mdn.getBytes());
            byte[] info = "cf1".getBytes();
            put.addColumn(info, "lg".getBytes(), start_time, longitude.getBytes());
            put.addColumn(info, "lat".getBytes(), start_time, latitude.getBytes());
            dianxin.put(put);

        }
    }

    // 获取用户最新10个位置
    // F88B07EB2778DA2B26EDFB9FE01EB4BEAD5C35A8
    // 如果只需要根据rowkey获取一条记录 那么使用get
    // 如果想获取多条记录 那么使用scan
    // 如果一条数据有多个版本 则需要使用CellUtil及每个Result对象的listCells方法一起获取
    @Test
    public void getLast10Loc() throws IOException {
        String rowkey = "F88B07EB2778DA2B26EDFB9FE01EB4BEAD5C35A8";
        System.out.println(rowkey + " ");
        Get get = new Get(rowkey.getBytes());
        // 设置获取多少个版本的数据
        get.setMaxVersions(10);
        Result rs = dianxin.get(get);
        List<String> latList = new ArrayList<>();
        List<String> lgList = new ArrayList<>();
        for (Cell cell : rs.listCells()) {
            String qf = Bytes.toString(CellUtil.cloneQualifier(cell));
            String value = Bytes.toString(CellUtil.cloneValue(cell));
            if ("lat".equals(qf)) {
                latList.add(value);
            } else if ("lg".equals(qf)) {
                lgList.add(value);
            }
        }
        for (int i = 0; i < 10; i++) {
            String lat = latList.get(i);
            String lg = lgList.get(i);
            System.out.println(lat + "," + lg);

        }

    }

    @After
    public void close() {
        try {
            admin.close();
            conn.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
